package com.gxkj.projects.myschoolx.binlog.v1;

import com.alibaba.fastjson.JSON;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.gxkj.projects.myschoolx.binlog.v1.model.CDCEvent;
import com.gxkj.projects.myschoolx.binlog.v1.model.ColumnInfo;
import com.gxkj.projects.myschoolx.binlog.v1.model.TableInfo;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class MyEventListener implements BinaryLogClient.EventListener {

    private static final Logger logger = LoggerFactory.getLogger(MyEventListener.class);

    private static String binlogFilename = null;

    private static long binlogPosition = 4;

    @Override
    public void onEvent(Event event) {
        if(event != null){

            EventHeaderV4 eventHeaderV4 = event.getHeader();

            if(event.getHeader().getEventType().equals(EventType.ROTATE)){
                RotateEventData data =  (RotateEventData)event.getData();
                binlogFilename = data.getBinlogFilename();
                binlogPosition = data.getBinlogPosition();

               // logger.info("事件：EventType.ROTATE，使用的binlog文件为："+binlogFilename+",binlogPosition为"+binlogPosition+"，建议做数据库记录binLogFileName 和 binlogPosition,以后注意nextPosition!=0时将binlogPosition更新为nextPosition");
            } else if(event.getHeader().getEventType().equals(EventType.TABLE_MAP)){
                /**
                 * 技巧及原理
                 * 每次ROW_EVENT前都伴随一个TABLE_MAP_EVENT事件，保存一些表信息，如tableId, tableName, databaseName, 而ROW_EVENT只有tableId
                 */
                TableMapEventData data  =  (TableMapEventData)event.getData();

                //缓存表信息
                MyTableInfoKeeper.saveTableIdMap(data.getTableId() ,data.getDatabase(),data.getTable());


            }  else if(event.getHeader().getEventType().equals(EventType.EXT_WRITE_ROWS)){


                WriteRowsEventData data  =  (WriteRowsEventData)event.getData();
                long tableId = data.getTableId();
                TableInfo tableInfo = MyTableInfoKeeper.getTableInfo(tableId);
                String databaseName = tableInfo.getDatabaseName();
                String tableName = tableInfo.getTableName();

                List<Serializable[]> rows = data.getRows();
                for(Serializable[] row: rows){

                    Map<String,String> afterMap = getMap(row,databaseName,tableName);

                    //生成解析内容并保存
                    CDCEvent cdcEvent = new CDCEvent(  binlogFilename,   databaseName,   tableName,
                            EventType.EXT_WRITE_ROWS, eventHeaderV4.getTimestamp(), eventHeaderV4.getServerId(),eventHeaderV4.getNextPosition());

                    cdcEvent.setAfter(afterMap);

                    cdcEvent.setIsDdl(false);

                    MyManager.queue.addLast(cdcEvent);
                }

            } else if(event.getHeader().getEventType().equals(EventType.EXT_UPDATE_ROWS)){

                UpdateRowsEventData  data  =  (UpdateRowsEventData)event.getData();
                long tableId = data.getTableId();
                TableInfo tableInfo = MyTableInfoKeeper.getTableInfo(tableId);
                String databaseName = tableInfo.getDatabaseName();
                String tableName = tableInfo.getTableName();

                List<Map.Entry<Serializable[], Serializable[]>> rows = data.getRows();
                for(Map.Entry<Serializable[], Serializable[]> p:rows){

                    Serializable[] before = p.getKey();
                    Serializable[] after  = p.getValue();

                    Map<String,String> beforeMap = getMap(before,databaseName,tableName);
                    Map<String,String> afterMap = getMap(after,databaseName,tableName);


                    //生成解析内容并保存
                    CDCEvent cdcEvent = new CDCEvent(  binlogFilename,   databaseName,   tableName,
                            EventType.EXT_UPDATE_ROWS, eventHeaderV4.getTimestamp(), eventHeaderV4.getServerId(),eventHeaderV4.getNextPosition());

                    cdcEvent.setAfter(afterMap);
                    cdcEvent.setBefore(beforeMap);

                    cdcEvent.setIsDdl(false);

                    MyManager.queue.addLast(cdcEvent);
                }

            } else if(event.getHeader().getEventType().equals(EventType.EXT_DELETE_ROWS)){

                DeleteRowsEventData  data  =  (DeleteRowsEventData)event.getData();
                long tableId = data.getTableId();
                TableInfo tableInfo = MyTableInfoKeeper.getTableInfo(tableId);
                String databaseName = tableInfo.getDatabaseName();
                String tableName = tableInfo.getTableName();

                List<Serializable[]> rows = data.getRows();

                for(Serializable[] row: rows){

                    Map<String,String> beforeMap = getMap(row,databaseName,tableName);

                    //生成解析内容并保存
                    CDCEvent cdcEvent = new CDCEvent(  binlogFilename,   databaseName,   tableName,
                            EventType.EXT_DELETE_ROWS, eventHeaderV4.getTimestamp(), eventHeaderV4.getServerId(),eventHeaderV4.getNextPosition());

                    cdcEvent.setBefore(beforeMap);

                    cdcEvent.setIsDdl(false);

                    MyManager.queue.addLast(cdcEvent);
                }
            } else if(event.getHeader().getEventType().equals(EventType.QUERY)){
                QueryEventData   data  =  (QueryEventData)event.getData();
                String databaseName = data.getDatabase();
                String sql = data.getSql();
                if(StringUtils.isEmpty(sql) || "BEGIN".equals(sql)){
                     //do nothing
                }else{
                    CDCEvent cdcEvent = new CDCEvent(  binlogFilename,   databaseName,   "",
                            EventType.QUERY, eventHeaderV4.getTimestamp(), eventHeaderV4.getServerId(),eventHeaderV4.getNextPosition());
                    cdcEvent.setIsDdl(true);
                    cdcEvent.setSql(sql);
                    MyManager.queue.addLast(cdcEvent);
                }
            }else{

                logger.info(JSON.toJSONString(event));
            }


        }
    }

    /**
     * 工具方法
     * ROW_EVENT中是没有Column信息的，需要通过MysqlConnection（下面会讲到）的方式读取列名信息，
     * 然后跟取回的List<Column>进行映射。
     * @param cols
     * @param databaseName
     * @param tableName
     * @return
     */
    private Map<String,String> getMap(Serializable[]  cols, String databaseName, String tableName){
        Map<String,String> map = new HashMap<>();
        if(cols == null || cols.length ==0){
            logger.info("cols="+(cols==null?"null":JSON.toJSONString(cols)));
            return null;
        }

        String fullName = databaseName+"."+tableName;
        List<ColumnInfo> columnInfoList = MyTableInfoKeeper.getColumns(fullName);
        if(columnInfoList == null){
            logger.info("没有查到表"+fullName+"的列信息");
            return null;
        }

        if(columnInfoList.size() != cols.length){
            //实时更新列记录
            MyTableInfoKeeper.refreshColumnsMap();
            if(columnInfoList.size() != cols.length)
            {
                logger.warn("columnInfoList.size is not equal to cols.");
                return null;
            }
        }

        for(int i=0;i<columnInfoList.size(); i++){
            if(cols[i]==null)
                map.put(columnInfoList.get(i).getName(),null);
            else
                map.put(columnInfoList.get(i).getName(), cols[i].toString());
        }

        return map;
    }
}
